package march.firstmarch22.march02;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.text.DecimalFormat;
import java.util.*;

public class Reducer02 extends Reducer< Text,Text,NullWritable,ScoreBean> {

    @Override
    protected void reduce( Text key,Iterable<Text> values, Context context) throws IOException, InterruptedException {

        //保存一位小数
        DecimalFormat df = new DecimalFormat("#0.0");
        HashMap<ScoreBean, Double> map = new HashMap<>();
        for (Text value : values) {
            String[] split = value.toString().split(",");
            ScoreBean scoreBean = new ScoreBean();

            String course = split[0];
            String name = split[1];

            scoreBean.setCourse(course);
            scoreBean.setName(name);

            //计算平均分
            double total = 0;
            int length = split.length;

            // 取出一条记录中的所有学生成绩，需要考虑split长度
            for (int i = length - 1; i > 1; i--) {
                total += Double.parseDouble(split[i]);
            }
            double avg = total/(length-2);
            avg = Double.parseDouble(df.format(avg));
            scoreBean.setAvg(avg);

            map.put(scoreBean,avg);

        }

        //进行排序
        List<Map.Entry<ScoreBean,Double>> list = new ArrayList<>(map.entrySet());
        list.sort(new Comparator<Map.Entry<ScoreBean, Double>>() {
            @Override
            public int compare(Map.Entry<ScoreBean, Double> o1, Map.Entry<ScoreBean, Double> o2) {
                if (o2.getValue() - o1.getValue() >= 0) {
                    return 1;
                } else {
                    return -1;
                }
            }
        });

        for (Map.Entry<ScoreBean, Double> entry : list) {

            context.write(NullWritable.get(),entry.getKey());
        }

    }
}

